package yh.jt808.jt808kafkaconsumer.kafka;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import yh.jt808.jt808kafkaconsumer.entity.CiVehicleAlarm;
import yh.jt808.jt808kafkaconsumer.entity.CiVehicleGps;
import yh.jt808.jt808kafkaconsumer.mapper.CiVehicleAlarmMapper;
import yh.jt808.jt808kafkaconsumer.mapper.CiVehicleGpsMapper;
import yh.jt808.jt808kafkaconsumer.vo.PositionAttribute;
import yh.jt808.jt808kafkaconsumer.vo.PositionReport;
import yh.jt808.jt808kafkaconsumer.websocket.WebSocketServer;

import javax.annotation.Resource;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;

/**
 * @author yanghx
 * @Date: 2020/6/12 23:28
 */
@Component
@Slf4j
public class KafkaConsumer {


    @Resource
    private WebSocketServer webSocketServer;
    @Resource
    private CiVehicleAlarmMapper ciVehicleAlarmMapper;
    @Resource
    private CiVehicleGpsMapper ciVehicleGpsMapper;

    /**
     * 消费监听
     *
     * @param data 数据
     */
    @KafkaListener(topics = {KafkaConstants.JT808_TOPIC})
    public void listen1(String data) {
        try {
            log.info("收到jt808地理位置数据 : {}", data);
            processResult(data);

        } catch (Exception e) {
            log.error("处理消息失败", e);
        }
    }

    private void processResult(String data) throws IOException {
        PositionReport positionReport = JSONUtil.toBean(data, PositionReport.class);
        List<PositionAttribute> attributes = positionReport.getPositionAttributes();

        //报警信息
        if (ObjectUtil.isNotEmpty(attributes)) {
            PositionAttribute positionAttribute = attributes.stream().filter(i -> 160 == i.getId()).findFirst().orElse(null);
            if (null != positionAttribute) {
                CiVehicleAlarm alarm = processAlarm(positionReport, positionAttribute);
                ciVehicleAlarmMapper.insert(alarm);
                return;
            }
        }
        //如果能走下来。就一定不是报警信息

        CiVehicleGps gps = processGps(positionReport);
        ciVehicleGpsMapper.insert(gps);
        webSocketServer.sendMessageOnAll(JSONUtil.toJsonStr(gps));
    }


    private CiVehicleGps processGps(PositionReport positionReport) {
        CiVehicleGps gps = new CiVehicleGps();
        BeanUtil.copyProperties(positionReport, gps);
        Integer latitude = positionReport.getLatitude();
        Integer longitude = positionReport.getLongitude();
        String lat = (latitude * 1.0F / 1000000) + "";
        String lon = (longitude * 1.0F / 1000000) + "";

        gps.setLatitude(lat);
        gps.setLongitude(lon);
        gps.setCreateTime(LocalDateTime.now());
        gps.setCreateUser("kafka");
        gps.setDeleteFlag(0);
        gps.setVehicleEquipmentCode(positionReport.getMobileNumber());
        return gps;
    }

    private CiVehicleAlarm processAlarm(PositionReport positionReport, PositionAttribute positionAttribute) {
        CiVehicleAlarm alarm = new CiVehicleAlarm();
        BeanUtil.copyProperties(positionReport, alarm);

        Integer latitude = positionReport.getLatitude();
        Integer longitude = positionReport.getLongitude();
        String lat = (latitude * 1.0F / 1000000) + "";
        String lon = (longitude * 1.0F / 1000000) + "";
        alarm.setLatitude(lat);
        alarm.setLongitude(lon);
        alarm.setCreateTime(LocalDateTime.now());
        alarm.setCreateUser("kafka");
        alarm.setDeleteFlag(0);
        alarm.setVehicleEquipmentCode(positionReport.getMobileNumber());
        alarm.setWarningBitValue(positionAttribute.getValueBit());

        return alarm;
    }
}



